package com.jasonchoi.redisqueue.queue;

import lombok.extern.log4j.Log4j2;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import java.util.Objects;
import java.util.concurrent.TimeUnit;

/**
 * REDIS -- 消息队列工具
 * @Author: JasonChoi
 * @Date: 2020/3/16 9:54
 */
@Log4j2
@Component
public class RedisQueueUtil {

    @Autowired
    private RedissonClient redissonClient;

    @Autowired
    private StringRedisTemplate redisTemplate;
    /**
     * 维护一个队列 防止消费失败
     */
    public static final String FAILED_TOPIC = "TEST_QUEUE:FAILED_MESSAGE";

    public <T> void publish(T t, String topic) {
        publish(t, 0L, TimeUnit.SECONDS, topic);
    }

    /**
     *  延迟消费 最好延迟5分钟以上，时间间隔太小容易消费失败
     * (延迟消费不稳定)
     * @param t
     * @param delay
     * @param timeUnit
     * @param topic
     * @param <T>
     */
    public <T> void publish(T t, long delay, TimeUnit timeUnit, String topic) {
        RBlockingQueue<T> blockingFairQueue = redissonClient.getBlockingQueue(topic);
        if (delay < 1) {
            log.debug("添加异步消息队列{}", topic);
            blockingFairQueue.offer(t);
            return;
        }
        log.debug("添加延时消息队列{},  delay:{},   timeUnit:{}", topic, delay, timeUnit);
        RDelayedQueue<T> delayedQueue = redissonClient.getDelayedQueue(blockingFairQueue);
        delayedQueue.offer(t, delay, timeUnit);
        delayedQueue.destroy();
    }

    public Object consume(String topic) throws InterruptedException {
        return redissonClient.getBlockingQueue(topic).take();
    }

    /**
     * 记录消费失败
     */
    public void recordFailed(String message, int count) {
        redisTemplate.opsForHash().put(FAILED_TOPIC, message, count + "");
    }

    public Integer getRetryCount(String message) {
        Object obj = redisTemplate.opsForHash().get(FAILED_TOPIC, message);
        return Objects.isNull(obj) ? null : Integer.valueOf(obj.toString());
    }

    public void removeFailed(String message) {
        redisTemplate.opsForHash().delete(FAILED_TOPIC, message);
    }
}
